/**
 * This file is part of the Kompics P2P Framework.
 *
 * Copyright (C) 2009 Swedish Institute of Computer Science (SICS)
 * Copyright (C) 2009 Royal Institute of Technology (KTH)
 *
 * Kompics is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; either version 2
 * of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
 */
package se.sics.drsm.components;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import se.sics.drsm.fd.ports.FailureDetector;
import se.sics.drsm.inits.*;
import se.sics.drsm.model.DRSMAddress;
import se.sics.drsm.model.DRSMConfiguration;
import se.sics.drsm.model.DRSMFingerTableView;
import se.sics.drsm.ports.drsm.DRSMPeriodicStabilizationPort;
import se.sics.drsm.ports.drsm.DRSMRouter;
import se.sics.drsm.ports.drsm.DRSMStructuredOverlay;
import se.sics.drsm.ports.drsm.events.*;
import se.sics.drsm.ports.statemachine.StateMachinePort;
import se.sics.kompics.*;
import se.sics.kompics.network.Network;
import se.sics.kompics.p2p.overlay.chord.ChordNeighborsRequest;
import se.sics.kompics.p2p.overlay.chord.ChordResponsibilityRequest;
import se.sics.kompics.p2p.overlay.chord.ChordStatus;
import se.sics.kompics.p2p.overlay.chord.LeaveRing;
import se.sics.kompics.timer.Timer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.LinkedList;

/**
 * The <code>DistributedRSM</code> class.
 *
 * @author Shadid Chowdhury <msrch@kth.se>
 */
public final class DistributedRSM extends ComponentDefinition {

    Negative<DRSMStructuredOverlay> son = negative(DRSMStructuredOverlay.class);
    Negative<ChordStatus> status = negative(ChordStatus.class);

    Positive<Network> network = positive(Network.class);
    Positive<Timer> timer = positive(Timer.class);
    Positive<FailureDetector> epfd = positive(FailureDetector.class);

    Component ring;
    Component router;
    Component stateMachine;
    Component communicationComponent;

    private int log2RingSize;
    private int successorListLength;
    private long stabilizationPeriod;
    private long fingerStabilizationPeriod;
    private long rpcTimeout;

    private boolean inside;
    private DRSMAddress self;
    //private DRSMAddress predecessor;
    private ArrayList<DRSMAddress> successorListView;
    private DRSMFingerTableView fingerTableView;
    private LinkedList<DRSMLookupRequest> earlyLookupRequests;

    private static Logger logger = LoggerFactory.getLogger(DistributedRSM.class);

    public DistributedRSM() {
        Class communicationClass = getCommunicationClass("drsm.properties");

        ring = create(DRSMNodePS.class);
        router = create(DRSMIterativeRouter.class);
        stateMachine = create(StateMachine.class);
        communicationComponent = create(communicationClass);


        connect(network, ring.getNegative(Network.class));
        connect(network, router.getNegative(Network.class));
        connect(network, stateMachine.getNegative(Network.class));
        connect(network, communicationComponent.getNegative(Network.class));

        connect(timer, ring.getNegative(Timer.class));
        connect(timer, router.getNegative(Timer.class));

        connect(epfd, ring.getNegative(FailureDetector.class));

        connect(ring.getNegative(DRSMRouter.class), router
                .getPositive(DRSMRouter.class));
        connect(router.getNegative(DRSMPeriodicStabilizationPort.class), ring
                .getPositive(DRSMPeriodicStabilizationPort.class));
        connect(ring.getNegative(StateMachinePort.class), stateMachine
                .getPositive(StateMachinePort.class));
        connect(router.getNegative(StateMachinePort.class), stateMachine
                .getPositive(StateMachinePort.class));
        connect(communicationComponent.getNegative(StateMachinePort.class), stateMachine
                .getPositive(StateMachinePort.class));

        subscribe(handleInit, control);
        subscribe(handleCreateRing, son);
        subscribe(handleJoinRing, son);
        subscribe(handleJoinRingCompleted, ring
                .getPositive(DRSMPeriodicStabilizationPort.class));
        subscribe(handleLeave, son);
        subscribe(handleLeaveCompleted, ring
                .getPositive(DRSMPeriodicStabilizationPort.class));
        subscribe(handleLookupRequest, son);
        subscribe(handleLookupResponse, router.getPositive(DRSMRouter.class));
        subscribe(handleResponsibilityRequest, son);
        subscribe(handleNeighborsRequest, status);
        subscribe(handleNewPredecessor, ring
                .getPositive(DRSMPeriodicStabilizationPort.class));
        subscribe(handleNewSuccessorList, ring
                .getPositive(DRSMPeriodicStabilizationPort.class));
        subscribe(handleNewFingerTable, router.getPositive(DRSMRouter.class));

        earlyLookupRequests = new LinkedList<DRSMLookupRequest>();
    }

    Handler<ChordInit> handleInit = new Handler<ChordInit>() {
        public void handle(ChordInit init) {
            inside = false;
            log2RingSize = init.getLog2RingSize();
            successorListLength = init.getSuccessorListLength();
            stabilizationPeriod = init.getStabilizationPeriod();
            fingerStabilizationPeriod = init.getFingerStabilizationPeriod();
            rpcTimeout = init.getRpcTimeout();

        }
    };

    Handler<DRSMCreateRing> handleCreateRing = new Handler<DRSMCreateRing>() {
        public void handle(DRSMCreateRing event) {
            logger.debug("handling DRSMCreateRing self {} ", event.getSelf());

            self = event.getSelf();
            logger.debug("{} Handling join request.", self);

            if (inside) {
                logger.error("Already inside ring");
                return;
            }

            // initialize ring
            trigger(new DRSMNodePSInit(log2RingSize, successorListLength,
                    stabilizationPeriod, self), ring.getControl());
            // initialize router
            trigger(new DRSMIterativeRouterInit(log2RingSize,
                    fingerStabilizationPeriod, rpcTimeout, self), router
                    .getControl());
            // initialize state machine
            trigger(new StateMachineInit(), stateMachine.getControl());
            // initialize communication component
            trigger(new GroupCommunicationInit(), communicationComponent.getControl());

            trigger(event, ring.getPositive(DRSMPeriodicStabilizationPort.class));
        }
    };

    Handler<DRSMJoinSuccessorGroup> handleJoinRing = new Handler<DRSMJoinSuccessorGroup>() {
        public void handle(DRSMJoinSuccessorGroup event) {
            logger.debug("handling DRSMJoinSuccessorGroup self {} ", event.getSelf());
            if (inside) {
                logger.error("Already inside ring");
                return;
            }
            self = event.getSelf();

            // initialize ring
            trigger(new DRSMNodePSInit(log2RingSize, successorListLength,
                    stabilizationPeriod, self), ring.getControl());
            // initialize router
            trigger(new DRSMIterativeRouterInit(log2RingSize,
                    fingerStabilizationPeriod, rpcTimeout, self), router
                    .getControl());
            // initialize state machine
            trigger(new StateMachineInit(), stateMachine.getControl());
            // initialize communication component
            trigger(new GroupCommunicationInit(), communicationComponent.getControl());

            trigger(event, ring.getPositive(DRSMPeriodicStabilizationPort.class));
        }
    };

    Handler<JoinRingCompleted> handleJoinRingCompleted = new Handler<JoinRingCompleted>() {
        public void handle(JoinRingCompleted event) {
            inside = true;
            trigger(event, son);
        }
    };

    Handler<LeaveRing> handleLeave = new Handler<LeaveRing>() {
        public void handle(LeaveRing event) {
            if (!inside) {
                if (logger != null) {
                    logger.error("Not inside ring");
                }
                return;
            }
            trigger(event, ring.getPositive(DRSMPeriodicStabilizationPort.class));
        }
    };

    Handler<LeaveRingCompleted> handleLeaveCompleted = new Handler<LeaveRingCompleted>() {
        public void handle(LeaveRingCompleted event) {
            inside = false;
            trigger(event, son);
        }
    };

    Handler<DRSMLookupRequest> handleLookupRequest = new Handler<DRSMLookupRequest>() {
        public void handle(DRSMLookupRequest event) {
            if (inside) {
                trigger(event, router.getPositive(DRSMRouter.class));
            } else {
                earlyLookupRequests.add(event);
            }
        }
    };

    Handler<DRSMLookupResponse> handleLookupResponse = new Handler<DRSMLookupResponse>() {
        public void handle(DRSMLookupResponse event) {
            trigger(event, son);
        }
    };

    Handler<ChordResponsibilityRequest> handleResponsibilityRequest = new Handler<ChordResponsibilityRequest>() {
        public void handle(ChordResponsibilityRequest event) {
//			ChordResponsibilityResponse response = new ChordResponsibilityResponse(
//					event, predecessor.getKey(), self.getKey());
//			trigger(response, son);
        }
    };

    Handler<ChordNeighborsRequest> handleNeighborsRequest = new Handler<ChordNeighborsRequest>() {
        public void handle(ChordNeighborsRequest event) {
//			ChordNeighborsResponse response = new ChordNeighborsResponse(event,
//					new ChordNeighbors(self, (successorListView == null ? self
//							: successorListView.get(0)), predecessor,
//							successorListView, fingerTableView));
//			trigger(response, status);
        }
    };

    private Handler<DRSMNewSuccessors> handleNewSuccessorList = new Handler<DRSMNewSuccessors>() {
        public void handle(DRSMNewSuccessors event) {
            logger.debug("NEW_SUCCESSOR_LIST {}", event.getSuccessorListView());

            successorListView = (ArrayList<DRSMAddress>) event.getSuccessorListView();
        }
    };

    private Handler<DRSMNewPredecessors> handleNewPredecessor = new Handler<DRSMNewPredecessors>() {
        public void handle(DRSMNewPredecessors event) {
//			logger.debug("NEW_PREDECESSOR {}", event.getPredecessorPeers());
//
//			if (predecessor != null && event.getPredecessorPeers() != null) {
//
//				if (!predecessor.equals(event.getPredecessorPeers())) {
//					predecessor = event.getPredecessorPeers();
//
//					trigger(new NewChordResponsibility(predecessor.getKey(),
//							self.getKey()), son);
//				}
//			} else if (event.getPredecessorPeers() != null
//					&& predecessor == null) {
//				predecessor = event.getPredecessorPeers();
//				trigger(new NewChordResponsibility(predecessor.getKey(), self
//						.getKey()), son);
//			}
        }
    };

    private Handler<CreateDRSMFingerTable> handleNewFingerTable = new Handler<CreateDRSMFingerTable>() {
        public void handle(CreateDRSMFingerTable event) {
            logger.debug("NEW_FINGER_TABLE {}", event.getFingerTableView());
            fingerTableView = event.getFingerTableView();

            if (inside && !earlyLookupRequests.isEmpty()) {
                for (DRSMLookupRequest request : earlyLookupRequests) {
                    trigger(request, router.getPositive(DRSMRouter.class));
                }
                earlyLookupRequests.clear();
            }

            trigger(event, son);
        }
    };


// Helper methods

    private Class getCommunicationClass(String configName) {
        DRSMConfiguration drsmConfiguration = getDRSMConfiguration(configName);
        String className = drsmConfiguration.getcommunicationClassName();
        Class classByName = getClassByName(className);

        return classByName;
    }

    // TODO: All the configs should be there in a static singleton class also make the methods generic
    private DRSMConfiguration getDRSMConfiguration(String configName) {
        DRSMConfiguration drsmConfiguration = null;

        try {
            drsmConfiguration = DRSMConfiguration.load(configName);
        } catch (IOException e) {
            e.printStackTrace();
            logger.error(e.getMessage());
            System.exit(1);
        }
        return drsmConfiguration;
    }

    private Class getClassByName(String className) {
        Class classByName = null;
        try {
            classByName = Class.forName(className);
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
            logger.error(e.getMessage());
        }
        return classByName;
    }
}

